package com.nokia.position;

import java.io.IOException;
//import java.util.ArrayList;
//import java.util.Collections;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class PositionServiceUsageUSA3Reducer extends
		Reducer<Text, PositionServiceUsageUSA3Writable, Text, Text> {

	//static final int PERCENTILE = 95;

	protected void reduce(Text key,
			Iterable<PositionServiceUsageUSA3Writable> values, Context context)
			throws IOException, InterruptedException {

		int totalServiceCnt = 0;
/*		
		int totalServiceSuccessCnt = 0;
		long totalHaccuracyCnt = 0;

		// um
		int mccSystemIdCnt = 0;
		int mncNetworkIdCnt = 0;
		int lacRncTacRzCnt = 0;
		int enbCnt = 0;
		int cellidBaseidCnt = 0;
		int neighborCellCnt = 0;
		int wlanCnt = 0;
		int referenceLocationCnt = 0;
		int wlanPlusNeighborCellCnt = 0;
		int wlanPlusCellIDBaseIDCnt = 0;
*/
		// mean, 95%
	//	ArrayList<Long> hAL = new ArrayList<Long>();

		for (PositionServiceUsageUSA3Writable value : values) {
			totalServiceCnt += value.getTotalCount();
/*
			totalServiceSuccessCnt += value.getStatusTrueCount();
			totalHaccuracyCnt += value.gethAccuracyCount();

			// um
			mccSystemIdCnt += value.getMccSystemId();
			mncNetworkIdCnt += value.getMNCNetworkId();
			lacRncTacRzCnt += value.getLacRncTacRz();
			enbCnt += value.getEnb();
			cellidBaseidCnt += value.getCellidBaseId();
			neighborCellCnt += value.getNeighborCell();
			wlanCnt += value.getWlan();
			referenceLocationCnt += value.getReferenceLocation();
			wlanPlusNeighborCellCnt += value.getWlanPlusNeighborCell();
			wlanPlusCellIDBaseIDCnt += value.getWlanPlusCellIDBaseID();

			// median, 95%
			hAL.add(value.gethAccuracyCount());
*/
		}

		// median
/*
		Collections.sort(hAL);
		long medianValue = 0;
		int N = hAL.size();
		if (hAL.size() % 2 == 0) {
			medianValue = (hAL.get(N / 2) + hAL.get(N / 2 - 1)) / 2;
		} else {
			medianValue = hAL.get((N - 1) / 2);
		}
*/
		// 95% linear interpolation closest ranks
/*		ArrayList<Integer> hPrAI = new ArrayList<Integer>();
		for (int n = 1; n <= N; n++) {
			hPrAI.add((int) (Math.round(100.0 / N * (n - 0.5))));
		}

		int t1 = 0;
		int tn1 = 0;
		int t2 = 0;
		int tn2 = 0;

		for (int i = N - 1; i > 0; i--) {
			if (hPrAI.get(i) > PERCENTILE) {
				t1 = hPrAI.get(i);
				tn1 = i;
				continue;
			} else if (t1 == PERCENTILE) {
				tn1 = i;
				tn2 = i;
				break;
			} else {
				t2 = hPrAI.get(i);
				tn2 = i;
				break;
			}
		}
		long pliValue = 0;
		if (tn1 == tn2) {
			pliValue = hAL.get(tn1);
		} else {
			pliValue = (long) (hAL.get(tn2) + ((N * (PERCENTILE - t2) / 100.0) * (hAL
					.get(tn1) - hAL.get(tn2))));
		}

		// 95% (nearest rank)
		// double nrank95Idx = hAL.size() * 95/100 + 0.5;
		// int nrank95IdxR = (int) Math.round(nrank95Idx);
		// long nrank95 = hAL.get(nrank95IdxR - 1);

		// ranging
		int hBidx = 0;
		final long[] hB = { 0, 100, 200, 250, 300, 400, 500, 600, 700, 800,
				900, 1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700, 1800,
				1900, 2000, 2100, 2200, 2300, 2400, 2500, 2600, 2700, 2800,
				2900, 3000 };
		long[] hBc = new long[hB.length];
		int hALidx;
		for (hALidx = 0; hALidx < hAL.size() && hBidx < hB.length;) {
			while (hBidx < hB.length) {
				if (hAL.get(hALidx) >= hB[hBidx]) {
					hBidx++;
				} else {
					hBc[hBidx - 1]++;
					hALidx++;
					break;
				}
			}
		}
		for (; hALidx < hAL.size(); hALidx++) {
			hBc[hBc.length - 1]++;
		}

		// For clarity on KPI-6 (QUA5) ranging. These sums can be calculated
		// from
		// hBc[]
		final long[] hB2 = { 0, 100, 250, 500, 2000 };
		long[] hBc2 = new long[hB2.length];
		hBc2[0] = hBc[0];
		hBc2[1] = hBc[1] + hBc[2];
		hBc2[2] = hBc[3] + hBc[4] + hBc[5];
		hBc2[3] = hBc[6] + hBc[7] + hBc[8] + hBc[9] + hBc[10] + hBc[11]
				+ hBc[12] + hBc[13] + hBc[14] + hBc[15] + hBc[16] + hBc[17]
				+ hBc[18] + hBc[19] + hBc[20];
		hBc2[4] = hBc[21] + hBc[22] + hBc[23] + hBc[24] + hBc[25] + hBc[26]
				+ hBc[27] + hBc[28] + hBc[29] + hBc[30] + hBc[31];
*/
		context.write(key, new Text("" + totalServiceCnt));
	}

}
